package com.gitee.sidihuo.utils.async;

import com.gitee.sidihuo.utils.util.AssertUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class SingleAsyncTask<P, R> {

    private static SingleAsyncTask instance = new SingleAsyncTask();

    private SingleAsyncTask() {
        try {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    log.info("SingleAsyncTask init start ... ");
                    executeTasks();
                }
            });
            thread.setName("SingleAsyncTask");
            thread.start();
        } catch (Throwable t) {
            log.warn("SingleAsyncTask init failed ", t);
        }
    }

    public static SingleAsyncTask getInstance() {
        return instance;
    }

    private final static int QUEUE_CAPACITY = 2000;
    private final static int QUEUE_CAPACITY_CLEAR = QUEUE_CAPACITY + 500;

    private LinkedBlockingQueue queue = new LinkedBlockingQueue(QUEUE_CAPACITY);
    private Map<String, AsyncWorkerFunction<P, R>> tasks = new ConcurrentHashMap<>();
    private Map<String, P> paras = new ConcurrentHashMap<>();
    private Map<String, SingleAsyncTaskContext> results = new ConcurrentHashMap<>();


    public boolean submitAsyncTask(SingleAsyncTaskContext context, AsyncWorkerFunction<P, R> workerFunction, P para) {
        String taskId = context.getTaskId();
        boolean offer = queue.offer(taskId);
        if (!offer) {
            log.warn("SingleAsyncTask submitAsyncTask failed {}", taskId);
        } else {
            tasks.put(taskId, workerFunction);
            paras.put(taskId, para);
            context.setTaskResult(null);
            results.put(taskId, context);
        }
        return offer;
    }

    public Object getAsyncTaskResult(SingleAsyncTaskContext context, int maxTimeWaitMillSeconds) {
        AssertUtils.assertContextParaLegal(context != null, "异步任务参数异常", "SingleAsyncTask getAsyncTaskResult context is null");
        AssertUtils.assertContextParaLegal(StringUtils.isNotBlank(context.getTaskId()), "异步任务参数taskId为空", "SingleAsyncTask getAsyncTaskResult context is empty");
        long currentTimeMillis = System.currentTimeMillis();
        Object result = null;
        while (true) {
            long currentTimeMillis2 = System.currentTimeMillis();
            long time = currentTimeMillis2 - currentTimeMillis;
            if (time > maxTimeWaitMillSeconds) {
                log.warn("SingleAsyncTask getAsyncTaskResult timeout {} -> {}", context.getTaskId(), time);
                break;
            }
            if (context.getTaskResult() != null) {
                result = context.getTaskResult();
                break;
            }
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                log.warn("SingleAsyncTask getAsyncTaskResult InterruptedException ", e);
            }
        }
        log.info("SingleAsyncTask getAsyncTaskResult finish {} {}", context.getTaskId(), System.currentTimeMillis() - currentTimeMillis);
        return result;
    }

    private void executeTasks() {
        while (true) {
            try {
                Object take = queue.take();
                if (!(take instanceof String)) {
                    continue;
                }
                AsyncWorkerFunction<P, R> workerFunction = tasks.get(take);
                if (workerFunction != null) {
                    tasks.remove(take);
                }
                P para = paras.get(take);
                if (para != null) {
                    paras.remove(take);
                }
                SingleAsyncTaskContext context = results.get(take);
                if (context != null) {
                    results.remove(take);
                }
                int sizeTasks = tasks.size();
                if (sizeTasks > QUEUE_CAPACITY_CLEAR || paras.size() > QUEUE_CAPACITY_CLEAR) {
                    tasks.clear();
                    paras.clear();
                    results.clear();
                    log.warn("SingleAsyncTask executeTasks while true 防止内存泄露  sizeTasks = {}", sizeTasks);
                }

                SingleAsyncTaskThreadPool.getThreadPool().execute(new Runnable() {
                    @Override
                    public void run() {
                        long currentTimeMillisB = System.currentTimeMillis();
                        String subTrace = (String) take;
                        log.info("SingleAsyncTask executeTasks execute begin sub thread subTrace = {}", subTrace);
                        R result = null;
                        try {
                            result = workerFunction.workAsync(para);
                        } catch (Exception e) {
                            log.warn("SingleAsyncTask executeTasks workerFunction workAsync failed", e);
                        }
                        context.setTaskResult(result);
                        long currentTimeMillisE = System.currentTimeMillis();
                        log.info("SingleAsyncTask executeTasks execute end sub thread subTrace = {} time = {}", subTrace, currentTimeMillisE - currentTimeMillisB);
                    }
                });
            } catch (Throwable e) {
                log.warn("SingleAsyncTask executeTasks while true failed ", e);
            }
        }
    }

}
